package mqttserver

import (
	"dev_gw/device"
	mqttclient "dev_gw/mqttClient"
	mqttmsg "dev_gw/mqttMsg"
	msgobj "dev_gw/msgObj"
	threadpool "dev_gw/threadPool"
	"fmt"
)

const defaultThreadNum = 10

type MqttServer struct {
	_pool   *threadpool.ThreadPool
	_mqtt   *mqttclient.MqttClient
	_device *device.Device

	_queue chan *mqttmsg.MqttMsg
}

func NewMqttServer(m *mqttclient.MqttClient, d *device.Device) *MqttServer {
	ms := &MqttServer{
		_pool:   threadpool.NewThreadPool(defaultThreadNum),
		_mqtt:   m,
		_device: d,
		_queue:  make(chan *mqttmsg.MqttMsg),
	}

	ms._mqtt.SetConnectCallback(func(clientId string) {
		fmt.Println(clientId, "connect success!!!")
		m.Subscribe("v1/dev/+/+/point/get")
		m.Subscribe("v1/app/+/dev_gw/point/get")
		m.Subscribe("v1/dev/+/dev_gw/point/set")
		m.Subscribe("v1/app/+/dev_gw/point/set")
	}).SetOnDisconnectCallback(func(s string) {
		fmt.Println(s, " connect lost!!!")
	}).SetOnMessageCallback(func(clientId, topic, payload string) {
		ms._queue <- mqttmsg.NewMqttMsg(topic, payload)
	}).SetPublishCallback(func(clientId, topic, payload string) {
		fmt.Println(clientId, "publish topic :", topic)
		fmt.Println(clientId, "publish payload :", payload)
	}).SetSubscribeCallback(func(clientId, topic string) {
		fmt.Println(clientId, "subscribe topic :", topic)
	}).SetUnsubscribeCallback(func(clientId, topic string) {
		fmt.Println(clientId, "unsubscribe topic :", topic)
	})

	ms._mqtt.Connect()

	return ms
}

func (ms *MqttServer) Start() {
	go func() {
		for {
			msg := <-ms._queue

			ms._pool.Submit(func() {
				fmt.Println("recv msg topic --> ", msg.Topic())
				fmt.Println("recv msg payload --> ", msg.Payload())

				msgObject := msgobj.Make(msg.Topic(), ms._mqtt, ms._device)
				if msgObject.Parse(msg.Topic(), msg.Payload()) {
					msgObject.Action().Reply()
				}
			})
		}
	}()
}
